package car_test;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class ccar {  // 保持原类名不变

    public static class MonthSalesMapper extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text month = new Text();  // 输出键：月份

        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            try {
                String line = new String(value.getBytes(), 0, value.getLength(), "GBK");
                String[] fields = line.trim().split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", -1);

                if (fields != null && fields.length >= 2) {  // 确保至少有月份列
                    String monthStr = fields[1].trim();  // 月份在第二列（索引1）

                    // 数据清洗：过滤无效月份
                    if (monthStr.isEmpty() || !monthStr.matches("^(0?[1-9]|1[0-2])$")) {
                        context.getCounter("DataError", "InvalidMonth").increment(1);
                        return;
                    }

                    // 标准化月份格式为两位（如"1"→"01"）
                    String normalizedMonth = String.format("%02d", Integer.parseInt(monthStr));
                    month.set(normalizedMonth);
                    context.write(month, one);  // 输出(月份, 1)
                }
            } catch (Exception e) {
                context.getCounter("DataError", "InvalidLine").increment(1);
            }
        }
    }

    public static class MonthSalesReducer extends Reducer<Text, IntWritable, Text, DoubleWritable> {
        private Map<String, Long> monthSales = new HashMap<>();  // 存储每月销量
        private long totalSales = 0;  // 全年总销量

        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            long sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            String month = key.toString();
            monthSales.put(month, sum);
            totalSales += sum;  // 累加总销量
        }

        @Override
        public void cleanup(Context context) throws IOException, InterruptedException {
            if (monthSales.isEmpty()) {
                context.write(new Text("统计提示"), new DoubleWritable(-1));
                return;
            }

            // 按月份排序输出（数值排序，非字典序）
            List<String> sortedMonths = monthSales.keySet().stream()
                    .sorted(Comparator.comparingInt(Integer::parseInt))
                    .collect(Collectors.toList());

            for (String month : sortedMonths) {
                long sales = monthSales.get(month);
                double ratio = (double) sales / totalSales;  // 计算月销售占比

                // 输出格式："01月-销量" 和 "01月-占比"
                context.write(new Text(month + "月-销量"), new DoubleWritable(sales));
                context.write(new Text(month + "月-占比"), new DoubleWritable(ratio));
            }
        }
    }

    public static void main(String[] args) throws Exception {
        // 主方法保持不变（仅调整输出目录）
        System.out.println("==================== 任务开始运行 ====================");
        System.out.println("开始时间：" + new java.util.Date().toString());

        String namenodeIp = "192.168.128.130";
        String hdfs = "hdfs://" + namenodeIp + ":8020";
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", hdfs);

        Job job = Job.getInstance(conf, "month_sales_analysis");
        job.setJarByClass(ccar.class);

        job.setMapperClass(MonthSalesMapper.class);
        job.setReducerClass(MonthSalesReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        String inputPath = hdfs + "/car/data.csv";
        String outputPath = hdfs + "/Output/month_only_v6";  // 输出目录调整

        FileInputFormat.addInputPath(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        FileSystem fs = FileSystem.get(conf);
        if (fs.exists(new Path(outputPath))) {
            fs.delete(new Path(outputPath), true);
        }

        boolean success = job.waitForCompletion(true);

        System.out.println("==================== 任务运行结束 ====================");
        System.out.println("结束时间：" + new java.util.Date().toString());
        System.out.println("任务状态：" + (success ? "成功" : "失败"));

        System.exit(success ? 0 : 1);
    }
}